package com.xpp.yanbi.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class MultiConsumer {

    private static final String TASK_QUEUE_NAME = "MULTI_QUEUE";

    public static void main(String[] argv) throws Exception {
        // 创建一个 ConnectionFactory 对象，用于配置和创建 RabbitMQ 连接
        ConnectionFactory factory = new ConnectionFactory();
        // 设置 RabbitMQ 服务器的主机为 localhost
        factory.setHost("localhost");

        // 创建到 RabbitMQ 服务器的连接
        final Connection connection = factory.newConnection();

        // 循环创建两个消费者
        for (int i = 0; i < 2; i++) {
            // 在连接上创建一个新的通道（Channel）
            final Channel channel = connection.createChannel();

            // 声明一个队列
            // 参数说明：
            // 1. TASK_QUEUE_NAME：队列的名称
            // 2. true：队列是否持久化（true 表示持久化，RabbitMQ 重启后队列仍然存在）
            // 3. false：队列是否独占（false 表示队列可以被多个连接共享）
            // 4. false：队列是否自动删除（false 表示队列不会在最后一个消费者断开后自动删除）
            // 5. null：队列的其他参数（例如消息的 TTL、队列的最大长度等）
            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

            // 打印提示信息，等待消息到来
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            // 设置消费者的 QoS（服务质量），每次只接收一条消息，处理完后再接收下一条
            // 这样可以确保消息的公平分发，避免某个消费者处理速度慢导致消息堆积
            channel.basicQos(1);

            // 定义消息处理逻辑
            int finalI = i; // 将循环变量 i 保存为 final 变量，以便在 Lambda 表达式中使用
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                // 将消息体转换为字符串
                String message = new String(delivery.getBody(), "UTF-8");

                try {
                    // 模拟处理消息的工作
                    System.out.println(" [x] Received '" + "编号:" + finalI + ":" + message + "'");

                    // 手动确认消息已被处理
                    // 参数说明：
                    // 1. delivery.getEnvelope().getDeliveryTag()：消息的唯一标识
                    // 2. false：是否批量确认（false 表示只确认当前消息,true 表示为一次性确认所有的历史消息直到当前这条）
                    // channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

                    // 暂停 20 秒，模拟处理消息需要一定时间
                    Thread.sleep(20000);
                } catch (InterruptedException e) {
                    // 如果处理过程中发生异常，打印异常信息
                    e.printStackTrace();

                    // 拒绝消息并重新入队
                    // 参数说明：
                    // 1. delivery.getEnvelope().getDeliveryTag()：消息的唯一标识
                    // 2. false：是否批量拒绝（false 表示只拒绝当前消息）
                    // 3. false：是否将消息重新放回队列（false 表示不重新入队）
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
                } finally {
                    // 无论是否发生异常，最终都会执行此代码块
                    System.out.println(" [x] Done");

                    // 再次确认消息已被处理（确保消息不会丢失）
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                }
            };

            // 开始消费消息
            // 参数说明：
            // 1. TASK_QUEUE_NAME：要消费的队列名称
            // 2. false：是否自动确认消息（false 表示手动确认）
            // 3. deliverCallback：消息处理回调函数
            // 4. consumerTag -> {}：消费者取消时的回调函数（这里为空实现）
            channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {});
        }
    }
}
